package com.shujia.flink.kafka;

import com.shujia.flink.core.Car;
import com.shujia.flink.core.Demo07Car;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Demo03Cars {
    public static void main(String[] args) throws Exception {

        // 构建Kafka的Source
        KafkaSource<String> kafkaSource = KafkaSource
                .<String>builder() // 指定Kafka中每条数据的格式
                .setBootstrapServers("master:9092,node1:9092,node2:9092") // 设置Kafka集群的地址
                .setTopics("cars") // 指定Topic
                .setGroupId("Demo03Cars") // 指定消费者组ID
                // 设置初始的偏移量 earliest：从最早开始消费  latest：从最新的消息开始消费
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定如何去解析Kafka中过来的每一条数据
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> carDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaCars");

        // 统计每条道路的平均车速以及车流量

        // 将每条数据转换成Car对象
        SingleOutputStreamOperator<Car> newCarDS = carDS.map(line -> {
            String[] splits = line.split(",");
            return new Car(splits[0], Integer.parseInt(splits[1]), Integer.parseInt(splits[2])
                    , splits[3], splits[4], splits[5]
                    , splits[6], Long.parseLong(splits[7]) * 1000, Double.parseDouble(splits[8]));
        });


        newCarDS.keyBy(car -> car.getRoadId())

                .process(new KeyedProcessFunction<String, Car, Result>() {
                    // 定义全局变量来保存车流量以及平均车速
                    // 如果全局变量定义在这里，那么Hash（roadId）% 并行度 取余的结果相同的道路会共用一套变量
                    // 需求是需要做到每条道路有自己单独的变量来保存结果
                    // 实际上是进入到同一个Task中的Key会共享一套全局变量
//                    Long carFlow = 0L;
//                    Double sumSpeed = 0.0;
//                    Double avgSpeed = 0.0;

                    // 以道路ID作为Key，需要保存的值作为Value，构建HashMap
                    HashMap<String, Long> carFlowMap = new HashMap<>();
                    HashMap<String, Double> sumSpeedMap = new HashMap<>();

                    /**
                     *
                     * @param car   DS中每一条数据
                     * @param ctx   上下文环境
                     * @param out   用于将结果输出到下游
                     */
                    // 每条数据会执行一次
                    @Override
                    public void processElement(Car car, KeyedProcessFunction<String, Car, Result>.Context ctx, Collector<Result> out) throws Exception {
                        String roadId = car.getRoadId();

                        // 从HashMap中通过roadId将之前的结果（之前的状态）取出来
                        Long carFlow = carFlowMap.getOrDefault(roadId, 0L);
                        carFlow++;
                        Double sumSpeed = sumSpeedMap.getOrDefault(roadId, 0.0);
                        sumSpeed += car.getSpeed();

                        // 将新结果（新的状态）写入HashMap中
                        carFlowMap.put(roadId, carFlow);
                        sumSpeedMap.put(roadId, sumSpeed);

                        // 计算平均速度
                        double avgSpeed = sumSpeed / carFlow;

                        out.collect(new Result(roadId, carFlow, avgSpeed));

                    }
                }).print();

        env.execute();


    }
}


class Result {
    private String roadId;
    private Long carCnt;
    private Double avgSpeed;

    public Result(String roadId, Long carCnt, Double avgSpeed) {
        this.roadId = roadId;
        this.carCnt = carCnt;
        this.avgSpeed = avgSpeed;
    }

    @Override
    public String toString() {
        return "Result{" +
                "roadId='" + roadId + '\'' +
                ", carCnt=" + carCnt +
                ", avgSpeed=" + avgSpeed +
                '}';
    }
}
